/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.ycl.st.load.worker;

import cn.ac.iie.di.ban.ycl.st.load.util.MongoHelper;
import cn.ac.iie.di.ban.ycl.st.load.util.SenderPool;
import cn.ac.iie.di.ban.ycl.st.load.util.StHelper;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.ConsumePosition;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;
import java.sql.Connection;
import org.apache.log4j.Logger;

/**
 *
 * @author zwlin
 */
public class Receiver {

    static private final Logger logger = Logger.getLogger(Receiver.class.getName());

    REReceiveSession session;
    ReceiveHandler handler;

    public Receiver(String receiveNamesrv, String receiveTopic, String groupnamePrefix, SenderPool[] senderPools) throws REConnectionException {
        Connection connection = StHelper.getConnection();
        handler = new ReceiveHandler(senderPools,connection);
        REConnection conn = new REConnection(receiveNamesrv);
        REReceiveSessionBuilder builder = (REReceiveSessionBuilder) conn.getReceiveSessionBuilder(receiveTopic);
        builder.setHandler(handler);
        builder.setFailureHandler(new REAbstractReceiveMessageHandler<byte[]>() {
            @Override
            public boolean handle(byte[] message) {
                logger.error("data format failed !");
                return true;
            }
        });
        if (groupnamePrefix == null || groupnamePrefix.trim().isEmpty()) {
            builder.setGroupName("iie_di_twx_st_test_" + receiveTopic);
        } else {
            builder.setGroupName(groupnamePrefix.trim() + receiveTopic);
        }
//        builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
         builder.setConsumPosition(ConsumePosition.CONSUME_FROM_FIRST_OFFSET);
        builder.setConsumeThreadNum(8);
        session = (REReceiveSession) builder.build();
    }

    public void start() throws REConnectionException {
        session.start();
    }
}
